package com.example.util.mq.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
public class MQManager {

    @Autowired
    private DisruptorEventHandler disruptorEventHandler;


    @Bean("disruptorMessageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {

        //指定事件工厂
        DisruptorEventFactory factory = new DisruptorEventFactory();

        //指定ringbuffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
        int bufferSize = 1024 * 256;

        //单线程模式，获取额外的性能
        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize,
                new ThreadFactory() {
                    private final AtomicInteger nextid = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable task) {
                        String name = "5G_threadFactory's say-hello-worker-" + nextid.getAndIncrement();
                        Thread thread = new Thread(task, name);
                        return thread;
                    }
                },
                ProducerType.SINGLE, new BlockingWaitStrategy());

        //设置事件业务处理器---消费者
        disruptor.handleEventsWith(disruptorEventHandler);

        // 启动disruptor线程
        disruptor.start();

        //获取ringbuffer环，用于接取生产者生产的事件
        RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();

        return ringBuffer;
    }
}